package uk.co.accolade.clash.multicast;

import uk.co.accolade.clash.core.ChangeHandlerFactory;
import uk.co.accolade.clash.core.UserChanges;
import uk.co.accolade.clash.core.message.ClashEventParseException;
import uk.co.accolade.clash.core.message.MessageParser;
import uk.co.accolade.clash.core.message.MessageParserFactory;
import uk.co.accolade.clash.core.message.MessageParser_V1;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class MulticastCommunicator implements Broadcaster, Receiver, Runnable {

    private static final Log LOG = LogFactory.getLog( MulticastCommunicator.class );

    public static final int SOCKET_PORT = 8001;
    public static final String GROUP_ADDRESS = "228.5.6.7";

    private MulticastSocket socket;
    private InetAddress group;
    private List<ChangeHandlerFactory> changeHandlerFactories;
    private boolean listen = false;

    public MulticastCommunicator(MulticastSocket socket, InetAddress group ) {
        this.socket = socket;
        this.group = group;
        this.changeHandlerFactories = new ArrayList<ChangeHandlerFactory>();
    }

    public void addChangeHandlerFactory( ChangeHandlerFactory changeHandlerFactory ) {
        this.changeHandlerFactories.add( changeHandlerFactory );
    }

    public void removeChangeHandlerFactory( ChangeHandlerFactory changeHandlerFactory ) {
        this.changeHandlerFactories.remove( changeHandlerFactory );
    }

    public void sendChange( UserChanges changes) throws IOException {
        String eventMessage = new MessageParser_V1().marshal( changes );                
        DatagramPacket data = new DatagramPacket(eventMessage.getBytes(), eventMessage.length(), group, SOCKET_PORT);
        socket.send(data);
    }

    public void register() throws IOException {
        socket.joinGroup( group );
    }

    public void deregister() throws IOException {
        socket.leaveGroup( group );
    }

    public void startListening() {
        Thread t = new Thread( this );
        t.start();
    }

    public void stopListening() {
        listen = false;        
    }

    public void handleMessage( String message ) {
        try {
            parseAndHandleChanges( message );
        } catch( ClashEventParseException ex ) {
            LOG.error( "problem parsing message [" + message + "] " + ex.getMessage(), ex );
        }
    }

    private void parseAndHandleChanges( String message ) {
        MessageParser parser = MessageParserFactory.getParserForMessage( message );
        UserChanges changes = parser.parse( message );
        for( ChangeHandlerFactory handlerFactory : changeHandlerFactories ) {
            handlerFactory.getRemoteChangesHandler().handleChanges( changes );
        }
    }

    public void run() {
        this.listen = true;
        byte[] buf = new byte[1000];
        DatagramPacket recv = new DatagramPacket(buf, buf.length);
        try {
            while( listen ) {
                LOG.info( "start listening" );
                socket.receive(recv);
                handleMessage( extractMessage( buf, recv ) );
            }
        } catch ( IOException e) {
            //todo log error and recover
            LOG.error( "exception while listening for messages: " + e.getMessage(), e );
            e.printStackTrace(); 
        }
    }

    private String extractMessage( byte[] buf, DatagramPacket recv ) {
        return new String( buf, 0, recv.getLength() );
    }
}
